RDD - Persistence Storage and levels

In Apache Spark, caching is not "one-size-fits-all". Depending on your cluster's hardware resources, memory capacity, and CPU speed, you may want to cache data differently.

For instance, if your RAM is highly constrained, you might choose to serialize data to save space, or spill overflow data to local worker disks instead of dropping it entirely.

Spark provides fine-grained control over caching via Storage Levels using the .persist() method. This guide details all available storage levels, their performance trade-offs, and provides complete PySpark examples.

Catalog of Spark Storage Levels

To use storage levels, you must first import the StorageLevel class from pyspark:

from pyspark import StorageLevel

Here is a detailed breakdown of all available levels:

Performance Trade-Off: Deserialized vs. Serialized

graph TD
    A["How to choose StorageLevel?"] --> B["1. Deserialized (MEMORY_ONLY)"]
    A --> C["2. Serialized (MEMORY_ONLY_SER)"]

    B --> B1["Pros: Blazing fast CPU access"]
    B --> B2["Cons: Massive RAM footprints, JVM GC pressure"]

    C --> C1["Pros: Highly compact RAM footprint, Low GC pauses"]
    C --> C2["Cons: Heavy CPU encoding/decoding overhead"]

    style A fill:#fff3e0,stroke:#e65100,stroke-width:2px;

PySpark Code Examples

Setup Spark Session

from pyspark.sql import SparkSession
from pyspark import StorageLevel

spark = SparkSession.builder \
    .appName("Day01 Persistence Levels") \
    .master("local[*]") \
    .getOrCreate()

sc = spark.sparkContext

A. Persisting with Memory and Disk spillover (MEMORY_AND_DISK)

This level keeps as much data in RAM as possible, and automatically spills overflow partitions to local disk instead of recomputing them:

# 1. Create a dummy dataset
data_rdd = sc.parallelize(range(10000), numSlices=2)

# 2. Persist with MEMORY AND DISK
data_rdd.persist(StorageLevel.MEMORY_AND_DISK)

# 3. Check if cached
print("Is RDD Cached?", data_rdd.is_cached) # True

# 4. Trigger an action to build the cache
count = data_rdd.count()
print(f"Total count: {count}")

# 5. Query the active storage level
level_info = data_rdd.getStorageLevel()
print("Active Storage Level Info:")
print(f"  Use Disk?      : {level_info.useDisk}")      # True
print(f"  Use Memory?    : {level_info.useMemory}")    # True
print(f"  Use OffHeap?   : {level_info.useOffHeap}")   # False
print(f"  Deserialized?  : {level_info.deserialized}") # True
print(f"  Replication No.: {level_info.replication}")   # 1

# 6. Unpersist to clear cache
data_rdd.unpersist()

B. Persisting with Serialization to save RAM (MEMORY_ONLY_SER)

Highly useful when dealing with millions of records to prevent running out of executor memory:

# 1. Create RDD
raw_logs = sc.parallelize(["ERROR: User timeout", "INFO: Job completed"] * 5000)

# 2. Persist in serialized byte form
serialized_rdd = raw_logs.persist(StorageLevel.MEMORY_ONLY_SER)

# 3. Trigger action to build serialized cache
total_logs = serialized_rdd.count()
print(f"Total Logs Cached: {total_logs}")

# 4. Verify storage level properties
info = serialized_rdd.getStorageLevel()
print("
--- Serialized Storage Level Check ---")
print("  Use Memory?   :", info.useMemory)    # True
print("  Use Disk?     :", info.useDisk)      # False
print("  Deserialized? :", info.deserialized) # False  (Serialized!)

# 5. Release cache
serialized_rdd.unpersist()

C. Persisting with High-Availability Replication (MEMORY_AND_DISK_2)

Use this level when running critical production pipelines where worker node crashes are common, and you cannot afford the latency of recomputing lost data:

# 1. Create RDD
critical_data = sc.parallelize(["Txn1,Approved", "Txn2,Failed"])

# 2. Persist with replication factor of 2 (writes to 2 executor nodes)
replicated_rdd = critical_data.persist(StorageLevel.MEMORY_AND_DISK_2)

# 3. Trigger action
replicated_rdd.count()

# 4. Verify replication count
info = replicated_rdd.getStorageLevel()
print("
--- Replicated Storage Level Check ---")
print("  Replication Factor:", info.replication) # 2

replicated_rdd.unpersist()

5. Crucial Storage Level Rules

1. Cannot change Storage Level on the Fly: Once you set a storage level on an RDD, you cannot change it without first releasing the cache. Attempting to call .persist() on an already cached RDD with a different level will throw a Runtime Error:

rdd = sc.parallelize([1, 2])
rdd.persist(StorageLevel.MEMORY_ONLY)
# This will FAIL:
# rdd.persist(StorageLevel.DISK ONLY)

1. How to change it: You must call .unpersist() first, and then apply .persist(new_level):

rdd.unpersist()
rdd.persist(StorageLevel.DISK_ONLY) # Success!